package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSqlApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd extends BaseSqlApp {

    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013,4,"DwdTradeCartAdd",Constant.TOPIC_DB);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table topic_db(\n" +
                "\t`database` String,\n" +
                "\t`data` MAP<string,string>,\n" +
                "\t`old` MAP<string,string>,\n" +
                "\t`type` String,\n" +
                "\t`table` String,\n" +
                "\t`ts` bigint,\n" +
                "  pt as proctime(),\n" +
                "  et as TO_TIMESTAMP_LTZ(ts, 0),\n" +
                "  WATERMARK FOR et AS et \n" +
                ")" + SqlUtil.getKafkaDDL("topic_db","DwdTradeCartAdd"));


        //tableEnv.sqlQuery("select * from topic_db where `table` = 'cart_info' and `type` = 'insert'").execute().print();

        Table table = tableEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n"+
                " `data`['course_id'] course_id,\n" +
                "\t`data`['sold'] sold,\n" +
                "\t`data`['cart_price'] cart_price,\n" +
                "\t`data`['user_id'] user_id,\n" +
                "\t`data`['course_name'] course_name,\n" +
                "\t`data`['session_id'] session_id\n" +
                " from topic_db\n" +
                " where `table` = 'cart_info'\n" +
                " and `type` = 'insert'");


        tableEnv.executeSql(" create table "+Constant.TOPIC_DWD_TRADE_CART_ADD+"(\n" +
                "\tid string,\n" +
                "\tcourse_id String,\n" +
                "\tsold String,\n" +
                "\tcart_price string,\n" +
                "\tuser_id string,\n" +
                "\tcourse_name string,\n" +
                "\tsession_id String,\n" +
                "\tPRIMARY KEY (id) NOT ENFORCED\n" +
                " ) "+SqlUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        table.executeInsert(Constant.TOPIC_DWD_TRADE_CART_ADD);

    }


}
